package com.caption.mq.config;

import com.alibaba.fastjson.JSONObject;
import com.caption.mq.annotation.MqConsumer;
import com.caption.mq.core.AbstractMqConsumer;
import com.caption.mq.core.RocketMqConsumer;
import com.caption.mq.entity.MqSubscribeDO;
import com.caption.mq.handler.MessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;

/**
 * @description: TODO
 * @author: 码农
 * @date: 2022-01-16 17:32:11
 */

@Slf4j
@Component
public class MqConsumerAutoConfiguration extends BaseMqAutoConfiguration {
    // 维护一份map用于检测是否用同样的consumerGroup订阅了不同的topic+tag
    private Map<String, String> validConsumerMap;


    public void init() throws Exception {

        Map<String, MessageHandler> beans = applicationContext.getBeansOfType(MessageHandler.class);
        if (CollectionUtils.isEmpty(beans)) {
            return;
        }
        validConsumerMap = new HashMap<>();
        for (Map.Entry<String, MessageHandler> entry : beans.entrySet()) {
            publishConsumer(entry.getKey(), entry.getValue());
        }
        //等待回收
        validConsumerMap = null;
    }

    private void publishConsumer(String beanName, MessageHandler messageHandler) throws MQClientException {
        if (StringUtils.isEmpty(mqProperties.getNameServer())) {
            throw new RuntimeException("name server address must be defined");
        }
        MqConsumer mqConsumer = applicationContext.findAnnotationOnBean(beanName, MqConsumer.class);
        Assert.notNull(mqConsumer.consumerGroup(), "consumer consumerGroup must be defined");
        Assert.notNull(mqConsumer.topic(), "consumer topic must be defined");
        String consumerGroup = mqConsumer.consumerGroup();

        if (!StringUtils.isEmpty(validConsumerMap.get(consumerGroup))) {
            String exist = validConsumerMap.get(consumerGroup);
            throw new RuntimeException("消费组重复订阅，请新增消费组用于新的topic和tag组合: " + consumerGroup + "已经订阅了" + exist);
        } else {
            validConsumerMap.put(consumerGroup, mqConsumer.topic() + "-" + mqConsumer.tag());
        }

        if (MessageHandler.class.isAssignableFrom(messageHandler.getClass())) {
            MqSubscribeDO mqSubscribe = getMqSubscribe(mqConsumer);
            AbstractMqConsumer consumer = new RocketMqConsumer(mqProperties.getNameServer(), mqSubscribe);
            consumer.subscribe(mqConsumer.topic(), mqConsumer.tag());
            consumer.registerMessageHandler(messageHandler);
            consumer.start();
            log.info("监听到消费的队列:{}", consumerGroup);
        }
    }

    /**
     * @param mqConsumer
     * @return
     */
    private MqSubscribeDO getMqSubscribe(MqConsumer mqConsumer) {


        MqSubscribeDO mqSubscribeDO = new MqSubscribeDO(mqConsumer.consumerGroup(), mqConsumer.topic(), mqConsumer.tag());
        mqSubscribeDO.setConsumeMode(mqConsumer.consumeMode());
        mqSubscribeDO.setMessageMode(mqConsumer.messageMode());
        mqSubscribeDO.setRetryCount(mqConsumer.retryCount());
        mqSubscribeDO.setRetryStrategy(mqConsumer.retryStrategy());
        if (mqProperties.getConsumeThreadMax() != null) {
            mqSubscribeDO.setConsumeThreadMax(mqProperties.getConsumeThreadMax());
        }
        if (mqProperties.getConsumeThreadMin() != null) {
            mqSubscribeDO.setConsumeThreadMin(mqProperties.getConsumeThreadMin());
        }
        if (mqProperties.getPullBatchSize() != null) {
            mqSubscribeDO.setPullBatchSize(mqProperties.getPullBatchSize());
        }
        if (mqProperties.getConsumeMessageBatchMaxSize() != null) {
            mqSubscribeDO.setConsumeMessageBatchMaxSize(mqProperties.getConsumeMessageBatchMaxSize());
        }

        return mqSubscribeDO;
    }

}
